package tasks.io.core.task;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tasks.io.client.ClientChannelContext;
import tasks.io.client.ReconnConf;
import tasks.io.client.TioClientConfig;
import tasks.io.core.ChannelContext;
import tasks.io.core.MaintainUtils;
import tasks.io.core.SystemTimer;
import tasks.io.core.thread.AbstractQueueTask;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;


public class CloseTask extends AbstractQueueTask<ChannelContext> {

    private static final Logger log = LoggerFactory.getLogger(CloseTask.class);

    public CloseTask(Executor executor) {
        super(executor);
        getQueue();
    }
    //	long count = 1;

    @Override
    public void runTask() {
        if (queue.isEmpty()) {
            return;
        }
        ChannelContext channelContext = null;
        while ((channelContext = queue.poll()) != null) {
            //			System.out.println(count++);
            try {
                boolean isNeedRemove = channelContext.closeMetaData.isNeedRemove;
                String remark = channelContext.closeMetaData.remark;
                Throwable throwable = channelContext.closeMetaData.throwable;


                channelContext.channelStat.timeClosed = SystemTimer.currTime;
                if (channelContext.tioConfig.getTioListener() != null) {
                    try {
                        channelContext.tioConfig.getTioListener().onBeforeClose(channelContext, throwable, remark, isNeedRemove);
                    } catch (Throwable e) {
                        log.error(e.toString(), e);
                    }
                }

                try {
                    //					channelContext.traceClient(ChannelAction.UNCONNECT, null, null);

                    if (channelContext.isClosed && !isNeedRemove) {
                        log.info("{}, {}已经关闭，备注:{}，异常:{}", channelContext.tioConfig, channelContext, remark, throwable == null ? "无" : throwable.toString());
                        return;
                    }

                    if (channelContext.isRemoved) {
                        log.info("{}, {}已经删除，备注:{}，异常:{}", channelContext.tioConfig, channelContext, remark, throwable == null ? "无" : throwable.toString());
                        return;
                    }

                    //必须先取消任务再清空队列
                    channelContext.decodeRunnable.setCanceled(true);
                    channelContext.handlerRunnable.setCanceled(true);
                    channelContext.sendRunnable.setCanceled(true);

                    channelContext.decodeRunnable.clearQueue();
                    channelContext.handlerRunnable.clearQueue();
                    channelContext.sendRunnable.clearQueue();

                    log.info("{}, {} 准备关闭连接, isNeedRemove:{}, {}", channelContext.tioConfig, channelContext, isNeedRemove, remark);

                    try {
                        if (isNeedRemove) {
                            MaintainUtils.remove(channelContext);
                        } else {
                            TioClientConfig tioClientConfig = (TioClientConfig) channelContext.tioConfig;
                            tioClientConfig.closeds.add(channelContext);
                            tioClientConfig.connecteds.remove(channelContext);
                            MaintainUtils.close(channelContext);
                        }

                        channelContext.setRemoved(isNeedRemove);
                        if (channelContext.tioConfig.statOn) {
                            channelContext.tioConfig.groupStat.closed.incrementAndGet();
                        }
                        channelContext.channelStat.timeClosed = SystemTimer.currTime;
                        channelContext.setClosed(true);
                    } catch (Throwable e) {
                        log.error(e.toString(), e);
                    } finally {
                        if (!isNeedRemove && channelContext.isClosed && !channelContext.isServer()) //不删除且没有连接上，则加到重连队列中
                        {
                            ClientChannelContext clientChannelContext = (ClientChannelContext) channelContext;
                            ReconnConf.put(clientChannelContext);
                        }
                    }
                } catch (Throwable e) {
                    log.error(throwable.toString(), e);
                }
                //				finally {
                //					writeLock.unlock();
                //				}
            } finally {
                channelContext.isWaitingClose = false;
            }
        }
    }


    /**
     * The msg queue.
     */
    private final ConcurrentLinkedQueue<ChannelContext> queue = new ConcurrentLinkedQueue<ChannelContext>();

    @Override
    public ConcurrentLinkedQueue<ChannelContext> getQueue() {

        return queue;
    }
}
